package transcoding

import controllers.LibraryManager
import com.xuggle.xuggler.IContainerFormat
import java.io.InputStream
import play.api.Logger
import java.util.concurrent.atomic.AtomicBoolean
import scala.actors.Actor
import java.io.BufferedReader
import java.io.InputStreamReader
import java.util.concurrent.ConcurrentHashMap
import java.util.UUID
import scala.collection.immutable.HashSet
import org.codehaus.jackson.annotate.{JsonProperty, JsonMethod, JsonAutoDetect}
import scala.Array


object TranscodeProcess {
  val ffmpeg = "ffmpeg"
  val video = ffmpeg + " -y -v 0 -ss 0 -i $file -async 1 -b:v $bitrate -s $size $audioSettings -f flv -vcodec $vcodec -preset superfast -threads 0 -"
  val transcodingCmds = Map(
    "video" -> (video),
    "video-alsowrite" -> (video + " -f $format $audioSettings -vcodec $vcodec -threads 0 -metadata grouping=$file -metadata comment=oktostream $outFile"),
    "video-justwrite" -> (ffmpeg + " -y -v 0 -ss 0 -i $file -f $format $audioSettings -vcodec $vcodec -threads 0 -metadata grouping=$file -metadata comment=oktostream $outFile"),
    "video-skipto" -> (ffmpeg + " -ss $offset -i $file -v 1 -async 1 -f flv -vcodec copy -acodec copy -preset superfast -threads 0 -"),
    "video-makeskippable" -> "qt-faststart $source $dest")

  /**
   * Creates a TranscodeProcess that when executed
   * will simply copy the file from the offset specified
   */
  def startSkipTo(mediaFile: MediaFile, offsetSeconds: Double)(implicit pi: ProcessInfo) = {
    // TODO: how to detect programmatically 
    // if the moov atom is at the beginning of the file?
    mediaFile.mediaType match {
      case v@Video =>
        transcodingCmds.get("video-skipto").map(
          cmd => cmd.replace("$offset", offsetSeconds.toString)
            .replace("$file", TranscoderUtil.encodeSpaces(mediaFile.filePath))) match {
          case None => throw new RuntimeException("Could not find relevant command for '" + v + "/skipto'")
          case Some(cmd) => new SkippedTranscodeStream(cmd)
        }
    }
  }

  def startRaw(mediaFile: MediaFile)(implicit pi: ProcessInfo) = {
    new CopyStreamTranscodeProcess(new java.io.FileInputStream(mediaFile.filePath))
  }

  def startTranscodingWithSave(mf: MediaFile, maxBitRate: Option[Int])(implicit pi: ProcessInfo) = {
    mf.mediaType match {
      case v@Video =>
        val extension = "mp4"
        val streamingFormat = TranscoderUtil.getXuggleFormat("flv")
        val outFile = mf.filePath.replace(LibraryManager.getLibraryDir, LibraryManager.getAltLibraryDir)
          .+("_bladeVersion." + extension + ".PART")

        transcodingCmds.get("video-alsowrite").map(
          cmd => {

            val vcodec = getVideoSettings(mf, streamingFormat, pi.player)
            val audioSettings = getAudioSettings(mf, streamingFormat, pi.player)

            cmd.replace("$file", TranscoderUtil.encodeSpaces(mf.filePath))
              .replace("$bitrate", maxBitRate.getOrElse(mf.bitRate.get).toString)
              .replace("$size", TranscoderUtil.size(maxBitRate)(mf))
              .replace("$audioSettings", audioSettings)
              .replace("$vcodec", vcodec)
              .replace("$format", extension)
              .replace("$outFile", TranscoderUtil.encodeSpaces(TranscoderUtil.ensureDirExists(outFile)))
          }) match {
          case None => throw new RuntimeException("Could not find relevant command for '" + v + "/alsowrite'")
          case Some(toExec) =>
            toExec
            new TranscodingAndSavingTranscodeProcess(toExec, mf, outFile)(new ProcessInfo(pi, Some(streamingFormat), maxBitRate))
        }
    }
  }

  def startTranscodingAdHoc(mf: MediaFile, maxBitRate: Option[Int])(implicit pi: ProcessInfo) = {
    mf.mediaType match {
      case v@Video =>
        val format = TranscoderUtil.getXuggleFormat("flv")
        transcodingCmds.get("video").map(
          cmd => {

            val vcodec = getVideoSettings(mf, format, pi.player)
            val audioSettings = getAudioSettings(mf, format, pi.player)

            cmd.replace("$file", TranscoderUtil.encodeSpaces(mf.filePath))
              .replace("$bitrate", maxBitRate.getOrElse(mf.bitRate.get).toString)
              .replace("$size", TranscoderUtil.size(maxBitRate)(mf))
              .replace("$audioSettings", audioSettings)
              .replace("$vcodec", vcodec)
          }) match {
          case None => throw new RuntimeException("Could not find relevant command for '" + v + "'")
          case Some(toExec) =>
            toExec
            new AdHocTranscodeProcess(toExec)(new ProcessInfo(pi, Some(format), maxBitRate))
        }
    }
  }

  def startTranscodingForSave(mf: MediaFile, maxBitRate: Option[Int])(implicit pi: ProcessInfo) = {
    mf.mediaType match {
      case v@Video =>
        val extension = "mp4"
        val format = TranscoderUtil.getXuggleFormat(extension)
        val isFileReady = pi.player.supports(mf)
        val outFile = if (isFileReady) mf.filePath
        else mf.filePath.replace(LibraryManager.getLibraryDir, LibraryManager.getAltLibraryDir)
          .+("_bladeVersion." + extension + ".PART")
        val destFilePath = if (isFileReady) mf.filePath + ".PART" else outFile.replaceAll("\\.PART$", "")

        // ffmpeg
        val cmd_1: (() => String, (java.lang.Process) => Unit) =
          if (isFileReady) null
          else {
            transcodingCmds.get("video-justwrite") match {

              case None => throw new RuntimeException("Could not find relevant command for '" + v + "/justwrite'")
              case Some(cmd) => {
                def toExecLater = {
                  val vcodec = getVideoSettings(mf, format, pi.player)
                  val audioSettings = getAudioSettings(mf, format, pi.player)

                  cmd.replace("$file", TranscoderUtil.encodeSpaces(mf.filePath))
                    .replace("$bitrate", maxBitRate.getOrElse(mf.bitRate.get).toString)
                    .replace("$size", TranscoderUtil.size(maxBitRate)(mf))
                    .replace("$audioSettings", audioSettings)
                    .replace("$vcodec", vcodec)
                    .replace("$format", extension)
                    .replace("$outFile", TranscoderUtil.encodeSpaces(TranscoderUtil.ensureDirExists(outFile)))
                }

                (toExecLater _, (p: java.lang.Process) => {
                  Logger.debug("executing on > " + Thread.currentThread.getName);
                  if (0 != p.waitFor)
                    throw new RuntimeException("'%s' returned with non-zero exit code" format TranscoderUtil.decodeSpaces(toExecLater))

                  if (!new java.io.File(outFile).exists)
                    throw new RuntimeException("Could not find expected transcoded file at " + outFile)
                })
              }
            }
          }

        // qt-faststart
        val cmd_2 = TranscodeProcess.transcodingCmds.get("video-makeskippable") match {

          case None => throw new RuntimeException("Could not find relevant command for '" + v + "/makeskippable'")
          case Some(cmd) => {
            def toExecLater = {
              cmd
                .replace("$source", TranscoderUtil.encodeSpaces(outFile))
                .replace("$dest", TranscoderUtil.encodeSpaces(destFilePath))
            }

            (toExecLater _, (p: java.lang.Process) => {
              Logger.debug("executing on > " + Thread.currentThread().getName());
              if (0 != p.waitFor)
                throw new RuntimeException("'%s' returned with non-zero exit code" format TranscoderUtil.decodeSpaces(toExecLater))

              val destinationFile = new java.io.File(destFilePath)
              val sourceFile = new java.io.File(outFile)

              if (isFileReady) {
                //destination might not exists if qt-faststart finds that the
                //video is already set up correctly
                if (destinationFile.exists && !destinationFile.renameTo(sourceFile))
                  throw new RuntimeException("Could not rename %s to  %s ".format(destFilePath, outFile))

              } else {
                if (!destinationFile.exists)
                  throw new RuntimeException("Could not find expected post-processed file at " + destFilePath)
                if (!sourceFile.delete)
                  Logger.warn("Could not clean up %" format outFile)

                Logger.debug("Registering saved file as alternative of " + mf.filePath)
                LibraryManager.addAlternativeFormat(mf, new java.io.File(destFilePath))
              }

            })
          }
        }

        Logger.debug("starting TranscodePipelineBackgroundProcess...")
        new TranscodePipelineBackgroundProcess(List(cmd_1, cmd_2).filter(tuple => null != tuple))
    }
  }


  private def getAudioSettings(mf: MediaFile, format: IContainerFormat, player: Player) = {
    if (!player.supportsAudio(mf.streamInfo(Audio), format))
      "-ar %d -ac 2".format(player.getNearestSupportedAudioSampleRate(mf.streamInfo(Audio).sampleRate.getOrElse(44100)))
    else
      "-acodec copy"
  }

  private def getVideoSettings(mf: MediaFile, format: IContainerFormat, player: Player) = {
    if (player.supportsVideo(mf.streamInfo(Video), format))
      "copy"
    else
      "libx264"
  }
}

abstract class TranscodeProcess(implicit val procInfo: ProcessInfo) {
  val gobbler = new StreamGobbler().start
  var pipeline = List[Map[String, Any]]()
  protected var started: Long = 0

  def startedTime = new java.util.Date(started)

  def isActive {}

  def getInputStream: TranscodedInputStream

  def getListener: InputStreamListener

  def isCancellable: Boolean

  //TODO:
  //def getStatus

  protected def newProcessBuilder(cmd: String) = {
    val argsPairs = new java.util.ArrayList[String]()
    cmd.split("\\s+").map(e => argsPairs add TranscoderUtil.decodeSpaces(e))

    new ProcessBuilder(argsPairs)
  }

  protected def startProcess(pb: ProcessBuilder) = {
    import collection.JavaConversions._
    Logger.debug("*** Running command: " + pb.command.foldLeft("")((z, b) => z + b + " "))

    val p = pb.start

    started = System.currentTimeMillis
    val procIS: InputStream = p.getInputStream()

    gobbler ! ('AddStream, p.getErrorStream())
    TranscodeProcessManager.registerProcessAsActive(this)

    p
  }

  protected def unregister() = TranscodeProcessManager.unregisterProcess(this)

  def destroy:Unit

  protected def destroy(p: java.lang.Process) = {
    try {
      TranscoderUtil.closeQuietly(p.getInputStream)
      TranscoderUtil.closeQuietly(p.getErrorStream)
      TranscoderUtil.closeQuietly(p.getOutputStream)

      p.destroy()
    }
    catch {
      case _ =>
    }
  }

  override def hashCode = {
    procInfo.PID.hashCode
  }

  override def equals(other: Any) = {
    if (other != null)
      other match {
        case o: TranscodeProcess =>
          procInfo.PID == procInfo.PID
        case _ => false
      }
    else false
  }
}

class TranscodePipelineBackgroundProcess(val pipeLineCmds: List[(() => String, java.lang.Process => Unit)])(implicit pi: ProcessInfo) extends TranscodeProcess {
  val active = new AtomicBoolean(true)
  val ps = new scala.collection.mutable.ListBuffer[java.lang.Process]

  val executor = new Actor {
    def act = {
      loop {
        react {
          case f: (() => Unit) => {
            f()
          }
          case _ => throw new RuntimeException("Incorrect message!")
        }
      }
    }
  }.start

  def toRun() = {
    Logger.debug("***toRun()");
    pipeLineCmds.foreach {
      cmd =>
        try {
          Logger.debug("***INSIDE toRun()");
          val p = startProcess(newProcessBuilder(cmd._1.apply))
          //also consume std output
          this.gobbler ! ('AddStream, p.getInputStream)
          ps += p
          cmd._2(p)//usually blocks on process
        } catch {
          case e => {
            destroy

            throw e
          }
        }
    }

    destroy
  }

  try {
    Logger.debug("***Executing from " + Thread.currentThread().getName());

    executor ! toRun _

  } catch {
    case _ =>
      destroy
  }

  override def isActive = active.get

  override def isCancellable = true

  override def destroy = {
    ps.foreach {
      p => destroy(p)
    }
    unregister()
    active.set(false)
  }

  override def getInputStream = null

  override def getListener = null
}

class TranscodingAndSavingTranscodeProcess(cmd: String, mf: MediaFile, savingFilePath: String)(implicit pi: ProcessInfo) extends SingleCmdTranscodeProcess(cmd) {

  private val destFilePath = savingFilePath.replaceAll("\\.PART$", "")

  trait CallBack extends Function1[Option[Throwable], Unit] {}

  class KeepAliveTranscodingProcess extends Actor {
    def act = {
      loop {
        react {
          // ((Option[Throwable]) => Unit))
          case ('KeepAlive, p: java.lang.Process, savingFilePath: String, callback: CallBack) =>

            Logger.info("[%s]Keeping alive transcoding process...".format(this.toString))
            Logger.info("Will postprocess " + savingFilePath)

            var reader: BufferedReader = null
            try {
              reader = new BufferedReader(new InputStreamReader(p.getInputStream))

              def read: Unit = {
                val line = reader.readLine()
                if (line != null)
                  read
              }
              read

              if (0 == p.waitFor()) {
                Logger.info("START post processing ...")

                val cmd = TranscodeProcess.transcodingCmds.get("video-makeskippable")
                  .getOrElse("")
                  .replace("$source", TranscoderUtil.encodeSpaces(savingFilePath))
                  .replace("$dest", TranscoderUtil.encodeSpaces(destFilePath))

                val postProcessCmd = startProcess(newProcessBuilder(cmd))
                if (0 == postProcessCmd.waitFor()) {
                  if (!new java.io.File(destFilePath).exists)
                    throw new RuntimeException("Could not find expected post-processed file at " + destFilePath)
                  if (!new java.io.File(savingFilePath).delete())
                    Logger.warn("Could not delete temp file " + savingFilePath)

                  Logger.info("... DONE post processing")
                } else throw new RuntimeException("PostProcessCmd (%s) returned a non-zero exit code".format(postProcessCmd))
              } else throw new RuntimeException("Transcoding process returned a non-zero exit code while dealing with file %s"
                .format(pi.mfm.mainMediaFile.filePath))

              callback(None)

            } catch {
              case error =>
                Logger.error("Error while keeping alive and registering saved transcoded file", error)
                callback(Some(error))
            } finally {
              TranscoderUtil.closeQuietly(reader)
              if (p != null) p.destroy
            }
          case msg => throw new RuntimeException("Incorrect message received %s".format(msg.toString))
        }
      }
    }
  }

  private val isInBackground = new AtomicBoolean(false)

  override def isCancellable = isInBackground.get

  override def getListener = new InputStreamListener {

    val onDone = new CallBack {
      override def apply(maybeError: Option[Throwable]) = {
        p.exitValue
        unregister
        maybeError match {
          case None =>
            Logger.debug("Registering saved file as alternative of " + mf.filePath)
            LibraryManager.addAlternativeFormat(mf, new java.io.File(destFilePath))

          case Some(error) => //TODO
        }
      }
    }

    override def onCloseRequest(is: InputStream) = {

      Logger.debug("Setting transcode process as cancellable but keeping it alive")
      isInBackground.set(true)

      val act = new KeepAliveTranscodingProcess().start
      act !('KeepAlive, p, savingFilePath, onDone)
    }
  }
}

class SkippedTranscodeStream(cmd: String)(implicit pi: ProcessInfo) extends SingleCmdTranscodeProcess(cmd)

abstract class SingleCmdTranscodeProcess(cmd: String)(implicit pi: ProcessInfo) extends TranscodeProcess {

  override def getListener = new InputStreamListener {
    override def onCloseRequest(is: InputStream) = {

      destroy
      unregister()
    }
  }
  val p = startProcess(newProcessBuilder(cmd))
  val is = new TranscodedInputStream(p.getInputStream, getListener)

  override def isCancellable = false

  override def getInputStream = is

  override def isActive = {
    try {
      if (p != null)
        p.exitValue // throws if active
      else false
    }
    catch {
      case _ => true
    }
  }

  override def destroy = destroy(p)
}

class AdHocTranscodeProcess(cmd: String)(implicit pi: ProcessInfo) extends SingleCmdTranscodeProcess(cmd)

class CopyStreamTranscodeProcess(val adapteeInputStream: InputStream)(implicit pi: ProcessInfo) extends TranscodeProcess {
  val isClosed = new AtomicBoolean(false)

  override def getListener = new InputStreamListener {
    override def onCloseRequest(is: InputStream) = {
      TranscoderUtil.closeQuietly(is)
      isClosed.set(true)
      unregister()
    }

    override def onSkip(skipped: Long) = {
      Logger.info("*** Skipped %d bytes" format skipped)
    }
  }

  val is = new TranscodedInputStream(adapteeInputStream, getListener)
  started = System.currentTimeMillis
  TranscodeProcessManager.registerProcessAsActive(this)

  override def isCancellable = false

  override def getInputStream = is

  override def isActive = {
    !isClosed.get
  }

  override def destroy = {
    try {
      adapteeInputStream.close
    }
    catch {
      case _ =>
    }
  }
}
@JsonAutoDetect(Array(JsonMethod.NONE))
class ProcessInfo(val PID: String, val destination: String, val mfm: MediaFileManager, val format: Option[IContainerFormat], val maxBitRate: Option[Int], val player: Player) {
  def this(PID: String, destination: String, mfm: MediaFileManager, player: Player) = this(PID, destination, mfm, None, None, player)

  def this(pi: ProcessInfo, format: Option[IContainerFormat], maxBitRate: Option[Int]) = this(pi.PID, pi.destination, pi.mfm, format, maxBitRate, pi.player)
  @JsonProperty
  def info: Map[String, String] =
    Map("PID" -> PID, "destination" -> destination, "format" -> format.getOrElse(mfm.mainMediaFile.format.get).getInputFormatLongName,
        "bitRate" -> maxBitRate.getOrElse(mfm.mainMediaFile.bitRate.getOrElse("N/A")).toString,
        "player" -> player.toString)

  override def hashCode = {
    (mfm.mainMediaFile.filePath + maxBitRate)
      .hashCode
  }

  override def equals(other: Any) = {
    other match {
      case o: ProcessInfo => {
        mfm.mainMediaFile.filePath == o.mfm.mainMediaFile.filePath && maxBitRate == o.maxBitRate
      }
      case _ => false
    }
  }
}

object TranscodeProcessManager {


  val activeProcessesMap = new ConcurrentHashMap[String, Set[TranscodeProcess]]

  def register(tp: TranscodeProcess, processMap: java.util.Map[String, Set[TranscodeProcess]]) {
    val key = tp.procInfo.mfm.mainMediaFile.filePath
    Logger.debug("KEY = " + key)

    processMap.put(key, key.synchronized(
    {
      ifNotNull(processMap.get(key)) {
        processesForFile => Some(processesForFile + tp)
      }.getOrElse(new HashSet + tp)
    }))
  }

  def unregister(tp: TranscodeProcess, processMap: java.util.Map[String, Set[TranscodeProcess]]) {
    val key = tp.procInfo.mfm.mainMediaFile.filePath
    Logger.debug("KEY = " + key)

    key.synchronized({
      ifNotNull(processMap.get(key)) {
        processesForFile => {
          val value = processesForFile - tp
          if (value.isEmpty) processMap.remove(key)
          else processMap.put(key, value)
          None
        }
      }
    })
  }

  /**
   * Tell the manager that <em>tp</em> process is now active
   */
  def registerProcessAsActive(tp: TranscodeProcess) = {

    register(tp, activeProcessesMap)

    Logger.debug("Registered " + tp)
  }

  /**
   * Tell the manager that <em>tp</em> process is now no longer active
   */
  def unregisterProcess(tp: TranscodeProcess) = {
    unregister(tp, activeProcessesMap)

    Logger.debug("Unregistered " + tp)
  }

  def findRegisteregProcesses(procInfo: ProcessInfo) = {
    val key = procInfo.mfm.mainMediaFile.filePath

    ifNotNull(activeProcessesMap.get(key)) {
      processesForFile =>
        processesForFile.find(p => p.procInfo == procInfo)
    }
  }

  def getProcessByPID(procInfo: ProcessInfo) = {
    val key = procInfo.mfm.mainMediaFile.filePath

    ifNotNull(activeProcessesMap.get(key)) {
      processesForFile =>
        processesForFile.find(p => p.procInfo.PID == procInfo.PID)
    }
  }

  def getTrancodingVideoStream(pi: ProcessInfo, mediaID: String, maxBitRate: Option[Int], format: Option[IContainerFormat], offset: Long) = {
    LibraryManager.getMediaLibrary.map(
      library => {
        ifNotNull(library.get(Video)) {
          libraryForType =>
            ifNotNull(libraryForType.get(mediaID)) {
              mfm => startProcessAndContructStream(pi, mfm, maxBitRate, format, offset)
            }
        }
      }).getOrElse(None)
  }

  def startInitialProcess(mfm: MediaFileManager) = {

    implicit val pi: ProcessInfo = new ProcessInfo(UUID.randomUUID.toString,
      "initial-process", mfm, mfm.mainMediaFile.format, mfm.mainMediaFile.bitRate, FlashPlayer)

    findRegisteregProcesses(pi) match {
      case Some(tp: TranscodePipelineBackgroundProcess) => Logger.debug("Found existing processing for %s" format mfm.mainMediaFile); tp
      case _ => TranscodeProcess.startTranscodingForSave(mfm.mainMediaFile, None)
    }
  }

  def startProcessAndContructStream(pi: ProcessInfo, mfm: MediaFileManager, maxBitRate: Option[Int], format: Option[IContainerFormat], offset: Double): Option[(InputStream, MediaFile)] = {

    implicit val procInfo = pi
    Logger.info("FORMAT REQUIRED " + format)
    Logger.info("ALTERNATIVE VERSIONS " + mfm.otherVersions.toString)
    mfm.getMediaFileWithMaxBitRateAndFormat(maxBitRate, format, pi.player) match {
      case Some(mf) => {
        Logger.info("FOUND FORMAT " + mf.format)
        // Matching Media File
        // => no transcoding required
        offset match {
          case 0 =>
            // No skipping requested
            // => send pure FileInputStream
            Logger.debug("TranscodeProcess.startRaw")
            Some((TranscodeProcess.startRaw(mf).getInputStream), mf)
          case secondOffset =>
            // Use ffmpeg with 'copy' encoder to skip
            // to the relevant portion of the stream
            // the only supported format if 'flv'
            Logger.debug("TranscodeProcess.startSkipTo with %.2f seconds offset" format secondOffset)
            Some((TranscodeProcess.startSkipTo(mf, secondOffset).getInputStream), mf)
        }
      }
      case None => {
        // Transcoding required, and if the bitrate/format 
        // is one of the standard one, we will track this
        // ffmpeg execution, which will also produce a regular file
        // and add this file to the alternative available to this MediaFileManager
        Logger.debug("TranscodeProcess.startTranscoding ...")
        findRegisteregProcesses(new ProcessInfo(pi, None, maxBitRate)) match {
          case None =>
            // to slow... the conversion is done either at upload time or in the background
            //Some(TranscodeProcess.startTranscodingWithSave(mfm.mainMediaFile, maxBitRate).getInputStream, mfm.mainMediaFile)
            Some((TranscodeProcess.startTranscodingAdHoc(mfm.mainMediaFile, maxBitRate).getInputStream, mfm.mainMediaFile))
          case Some(existingProcInfo) =>
            Logger.debug("Found existing TranscodingWithSave process, so just transcoding..." + existingProcInfo)
            Some((TranscodeProcess.startTranscodingAdHoc(mfm.mainMediaFile, maxBitRate).getInputStream, mfm.mainMediaFile))
        }
      }
    }
  }

  def ifNotNull[A, B](value: B)(f: B => Option[A]) = {
    if (null == value) None
    else f(value)
  }
}
